package com.qssoft.cqry.synchdb.handler;

import cn.hutool.crypto.digest.MD5;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.qssoft.cqry.synchdb.properties.SynchDbProperties;
import com.qssoft.cqry.synchdb.vo.MysqlBinlogVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author liju.z
 */
public class RedisSynchHandler {

    private RedisTemplate<String, Object> redisTemplate;

    private final String serverKey = "synch_mysql_db_";

    private final MD5 md5 = MD5.create();

    @Autowired
    private SynchDbProperties.SynchMysqlProperties synchMysqlProperties;

    public RedisSynchHandler(SynchDbProperties.SynchMysqlProperties synchMysqlProperties, RedisTemplate<String, Object> redisTemplate){
        this.synchMysqlProperties = synchMysqlProperties;
        this.redisTemplate = redisTemplate;
    }


    /**
     * 客服端推消息
     *
     * @param mysqlBinlogVo
     */
    public void sendSynchData(MysqlBinlogVo mysqlBinlogVo) {
        String mapKey = md5.digestHex(mysqlBinlogVo.getQueryEventData().getSql());
        redisTemplate.opsForValue().set(serverKey + synchMysqlProperties.getServerId() + ":" + mapKey, mysqlBinlogVo, 1, TimeUnit.HOURS);
    }
    /**
     * 服务端拉取消息
     *
     * @return
     */
    public List<MysqlBinlogVo> getSynchData() {
        return redisTemplate.execute((RedisCallback<List<MysqlBinlogVo>>) connection -> {
            List<MysqlBinlogVo> mysqlBinlogVoList = new ArrayList<>();
            Set<byte[]> keySets = connection.keys((serverKey + "*").getBytes());
            FastJsonRedisSerializer<MysqlBinlogVo> fastJsonRedisSerializer = new FastJsonRedisSerializer<>(MysqlBinlogVo.class);
            for (byte[] keySet : keySets) {
                String key = StringRedisSerializer.UTF_8.deserialize(keySet);
                //不是同步值不要
                if (!key.startsWith(serverKey)) continue;
                    //本工程发送的不处理
                else if (key.startsWith(serverKey + synchMysqlProperties.getServerId())) continue;

                byte[] value = connection.get(keySet);
                if (value == null) continue;
                MysqlBinlogVo mysqlBinlogVo = fastJsonRedisSerializer.deserialize(value);
                mysqlBinlogVoList.add(mysqlBinlogVo);
            }
            return mysqlBinlogVoList;
        });
    }

    public void sendMapData(MysqlBinlogVo mysqlBinlogVo){
        String key = serverKey + synchMysqlProperties.getServerId() + ":" + mysqlBinlogVo.getApplicationName();
        String mapKey = MD5.create().digestHex(mysqlBinlogVo.getQueryEventData().getSql());
        redisTemplate.opsForHash().put(key, mapKey, mysqlBinlogVo);
    }

    public void deleteSynchData(MysqlBinlogVo mysqlBinlogVo) {
        String mapKey = md5.digestHex(mysqlBinlogVo.getQueryEventData().getSql());
        redisTemplate.delete(serverKey + synchMysqlProperties.getServerId() + ":" + mapKey);
    }


}
